package top.callback.inf.beautyleaf.segment;

import lombok.extern.slf4j.Slf4j;
import top.callback.inf.beautyleaf.IDGenerator;
import top.callback.inf.beautyleaf.common.IDResult;
import top.callback.inf.beautyleaf.common.thread.MonitoredRunnable;
import top.callback.inf.beautyleaf.common.thread.MonitoredThreadPool;
import top.callback.inf.beautyleaf.common.thread.NamedThreadFactory;
import top.callback.inf.beautyleaf.common.utils.MybatisHelper;
import top.callback.inf.beautyleaf.segment.dao.LeafTagDao;
import top.callback.inf.beautyleaf.segment.model.LeafTag;
import top.callback.inf.beautyleaf.segment.model.Segment;
import top.callback.inf.beautyleaf.segment.model.SegmentBuffer;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 基于数据库号码段算法实现的 ID 生成器。号码段算法将两个号码段 {@link Segment} 组成一个缓冲区 {@link SegmentBuffer}，
 * 通过缓冲区来不断切换使用两个号码段实现更小的数据库压力和更高效率的 ID 生成。
 *
 * @author Pengyu Gan
 * CreateDate 2021/3/23
 */
@Slf4j
public class SegmentIDGenerator implements IDGenerator {

    private LeafTagDao leafTagDao;

    /**
     * 缓冲区缓存，key 为标签名称(tagName)，value 为对应的缓冲区
     */
    private final ConcurrentHashMap<String, SegmentBuffer> bufferCache = new ConcurrentHashMap<>();

    /**
     * 一个号码段的理论发号周期，这个值表示理论上一个号码段中的号码应该在这个时间跨度内发完。
     * 如果两次申请号码段的时间跨度
     *      小于这个值，表示发号速度比较快，那么下次就会尝试申请更多(2倍)的号码；
     *      大于这个值但小于两倍这个值，表示发号速度不快不慢，那么下次正常申请号码；
     *      大于两倍这个值，表示发号速度比较慢，那么下次就会尝试申请较少(一半)的号码。
     */
    private static final int SEGMENT_DURATION = 15 * 60 * 1000;

    /**
     * 每次申请号码段时允许的最大步长
     */
    private static final int MAX_STEP = 100000;

    /**
     * CPU 核心数
     */
    private final int cpuCores = Runtime.getRuntime().availableProcessors();

    /**
     * 加载下一个号码段任务的线程池。在号码段缓冲区内存在两个号码段交替使用，全部满足下面条件时，另一个当前没有使用的号码段将进行号码申请。
     *      1. 如果另一个当前没有使用的号码段没有准备好；
     *      2. 当前号码段已经发号超过 10%；
     *      3. 另一个当前没有使用的号码段没有正在加载号码
     */
    private final ExecutorService loadNextSegmentExecutorService = new MonitoredThreadPool(
            5,
            cpuCores * 2,
            60L,
            TimeUnit.SECONDS,
            new SynchronousQueue<>(),
            "xxxxx");

    /**
     * ID 生成器是否已初始化标识，
     */
    private volatile boolean init = false;

    /**
     * 确保缓存和数据库的标签数据一致 定时任务的执行周期
     */
    private int keepConsistencePeriod;

    /**
     * 加载另一个号码段的阈值
     */
    private double loadSegmentThreshold;

    /**
     * Mybatis 辅助工具
     */
    private MybatisHelper helper;

    public SegmentIDGenerator(int keepConsistencePeriod, double loadSegmentThreshold, MybatisHelper helper) {
        this.keepConsistencePeriod = keepConsistencePeriod;
        this.loadSegmentThreshold = loadSegmentThreshold;
        this.helper = helper;
    }

    @Override
    public boolean init() {
        leafTagDao = new LeafTagDao(helper);
        // 先把数据库里的标签加载到缓存
        keepCacheAndDBConsistent();
        init = true;
        // 启动定时任务定时同步
        periodicalKeepCacheAndDBConsistent();
        return init;
    }

    @Override
    public IDResult get(String tagName) {
        if (!init) {
            return IDResult.error(IDResult.SEGMENT_BUFFER_NOT_INIT);
        }
        if (!bufferCache.containsKey(tagName)) {
            return IDResult.error(IDResult.TAG_NOT_EXISTS);
        }
        SegmentBuffer buffer = bufferCache.get(tagName);
        if (!buffer.isBufferInitOk()) {
            synchronized (buffer) {
                if (!buffer.isBufferInitOk()) {
                    try {
                        loadSegment(tagName, buffer.getCurrent());
                        buffer.setBufferInitOk(true);
                    } catch (Exception e) {
                        log.error("标签 {} 初始化缓冲区异常", tagName, e);
                    }
                }
            }
        }
        return getId(buffer);
    }

    /**
     * 定时调度任务确保缓存和数据库的标签数据一致
     */
    private void periodicalKeepCacheAndDBConsistent() {
        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("keepCacheAndDBConsistent"));
        scheduler.scheduleWithFixedDelay(new MonitoredRunnable(
                this::keepCacheAndDBConsistent,
                t -> log.error("定时任务 确保缓存和数据库的标签数据一致 时发生异常", t)),
                keepConsistencePeriod, keepConsistencePeriod, TimeUnit.SECONDS);
    }

    /**
     * 确保缓存和数据库的标签数据一致。方法每次执行先将数据库中新增而缓存中没有的标签加载到缓存，
     * 再将数据库中已删除而缓存中还存在的标签删除。
     */
    private void keepCacheAndDBConsistent() {
        try {
            // 查询数据库中所有的标签名称
            List<String> persistentTagNames = leafTagDao.findAllTagNames();
            if (null == persistentTagNames || persistentTagNames.isEmpty()) {
                throw new IllegalStateException("数据库中不存在任何标签，无法启动。");
            }
            // 当前缓存中已存在的标签名称
            List<String> cachedTagNames = new ArrayList<>(bufferCache.keySet());
            // 准备添加到缓存的标签名称
            Set<String> preAddTagNames = new HashSet<>(persistentTagNames);
            // 删除缓存中已存在的标签名称
            for (String cachedTagName : cachedTagNames) {
                preAddTagNames.remove(cachedTagName);
            }
            // 根据标签名称创建缓冲区
            for (String preAddTagName : preAddTagNames) {
                SegmentBuffer buffer = new SegmentBuffer();
                buffer.setTagName(preAddTagName);
                Segment segment = buffer.getCurrent();
                segment.setValue(new AtomicLong(0));
                segment.setMax(0);
                segment.setStep(0);
                bufferCache.put(preAddTagName, buffer);
                log.info("标签 {} 被加载到缓存，对应缓冲区信息 {}", preAddTagName, buffer);
            }
            // 准备从缓存删除的标签名称
            Set<String> preRemoveTagNames = new HashSet<>(cachedTagNames);
            // 删除缓存中存在、但是在数据库中已经不存在的标签名称
            for (String persistentTagName : persistentTagNames) {
                preRemoveTagNames.remove(persistentTagName);
            }
            for (String preRemoveTagName : preRemoveTagNames) {
                bufferCache.remove(preRemoveTagName);
                log.info("标签 {} 从缓存中被删除", preRemoveTagName);
            }
        } catch (Exception e) {
            log.error("确保缓存和数据库的标签数据一致时发生异常。", e);
        }
    }

    /**
     * 为指定号码段加载 ID
     * @param tagName 标签名称
     * @param segment 要加载 ID 的号码段
     */
    private void loadSegment(String tagName, Segment segment) {
        // 号码段所在缓冲区
        SegmentBuffer buffer = segment.getBuffer();
        LeafTag leafTag;
        // 缓冲区没有初始化，表示启动后第一次操作
        if (!buffer.isBufferInitOk()) {
            // 上次启动申请的号码段中的号码可能没有用完，为了避免发出同样的号码，这里直接申请 [maxId, maxId+step] 这段号码
            leafTag = leafTagDao.updateMaxIdAndGet(tagName);
            buffer.setStep(leafTag.getStep());
            buffer.setMinStep(leafTag.getStep());
        } else if (buffer.getLastUpdateTimestamp() == 0) { // updateTimestamp 为 0 表示这是启动后的第一次更新(即第二次操作)
            leafTag = leafTagDao.updateMaxIdAndGet(tagName);
            buffer.setLastUpdateTimestamp(System.currentTimeMillis());
            buffer.setStep(leafTag.getStep());
            buffer.setMinStep(leafTag.getStep());
        } else { // 第 N(N > 2) 次操作
            // 缓冲区上一次申请号码段距离当前时间点的跨度
            long duration = System.currentTimeMillis() - buffer.getLastUpdateTimestamp();
            // 计算本次申请号码段所使用的步长
            int nextStep = buffer.getStep();
            // 时间跨度超过了号码段的持续时间，意味着发号速度比较快，尝试申请两倍步长的号码段
            if (duration < SEGMENT_DURATION) {
                // 如果 2 倍步长超过了最大允许的步长，则申请号码段时不调整步长
                if (nextStep * 2 > MAX_STEP) {
                    // do nothing
                } else { // 否则申请号码段时使用两倍步长
                    nextStep = nextStep * 2;
                }
            } else if (duration < SEGMENT_DURATION * 2) {
                // 发号速度不快不慢，维持现状
            } else { // 发号速度比较慢，尝试申请一半步长的号码段
                nextStep = nextStep / 2 >= buffer.getMinStep() ? nextStep / 2 : nextStep;
            }
            LeafTag temp = new LeafTag();
            temp.setTagName(tagName);
            temp.setStep(nextStep);
            // 确定了步长后执行号码段申请
            leafTag = leafTagDao.updateMaxIdBySpecificStepAndGet(temp);
            // 更新 buffer 的变更时间
            buffer.setLastUpdateTimestamp(System.currentTimeMillis());
            // 设置本次调整后的步长
            buffer.setStep(nextStep);
            // 最小步长取标签信息上的步长
            buffer.setMinStep(leafTag.getStep());
        }
        // 号码段起始号码 = 最大 ID - 计算后的步长
        long value = leafTag.getMaxId() - buffer.getStep();
        segment.getValue().set(value);
        segment.setMax(leafTag.getMaxId());
        segment.setStep(buffer.getStep());
    }

    /**
     * 获取一个 ID。先尝试从当前正在使用的号码段获取，获取之前先判断是否需要加载另一个号码段。如果当前号码段
     * 没有 ID 了则等另一个号码段加载完成后切换到新号码段获取。如果另一个号码段加载失败，当前号码段也没有了 ID
     * 则发号失败。
     * @param buffer 号码段缓冲区
     * @return ID
     */
    public IDResult getId(SegmentBuffer buffer) {
        while (true) {
            buffer.rLock().lock();
            try {
                // 拿到当前正在使用的号码段
                Segment segment = buffer.getCurrent();
                // 满足下列要求，对另一个没使用的号码段进行号码加载：
                // 1. 如果另一个当前没有使用的号码段没有准备好；
                // 2. 当前号码段已经发号超过 10%；
                // 3. 另一个当前没有使用的号码段没有正在加载号码
                if (!buffer.isNextSegmentReady() && (segment.getIdle() < loadSegmentThreshold * segment.getStep()) && buffer.getLoadSegmentThreadRunning().compareAndSet(false, true)) {
                    loadNextSegmentExecutorService.execute(new MonitoredRunnable(() -> {
                        // 另一个未使用的号码段
                        Segment next = buffer.getSegments()[buffer.nextSegmentPosition()];
                        // 加载完成标识符
                        boolean loadOk = false;
                        try {
                            // 加载号码段
                            loadSegment(buffer.getTagName(), next);
                            // 加载完成
                            loadOk = true;
                        } catch (Exception e) {
                            log.error("标签 {} 加载号码段异常", buffer.getTagName(), e);
                        } finally {
                            if (loadOk) { // 加载完成设置标志位
                                buffer.wLock().lock();
                                buffer.setNextSegmentReady(true);
                                buffer.getLoadSegmentThreadRunning().set(false);
                                buffer.wLock().unlock();
                            } else { // 加载失败
                                buffer.getLoadSegmentThreadRunning().set(false);
                            }
                        }
                    }, t -> log.error("标签 {} 的缓冲区加载另一块号码段时异常。", buffer.getTagName(), t)));
                }
                // 上面这个 if 块内是异步任务，这里是从当前使用的号码段里取到下一个号码
                long value = segment.getValue().getAndIncrement();
                // 如果取到的号码小于号码段的最大号码则取号成功并返回
                if (value < segment.getMax()) {
                    return IDResult.ok(value);
                }
            } finally {
                buffer.rLock().unlock();
            }
            // 自旋等待另一个号码段加载完成
            waitAsyncLoadSegmentTaskFinish(buffer);
            buffer.wLock().lock();
            try {
                Segment segment = buffer.getCurrent();
                long value = segment.getValue().getAndIncrement();
                if (value < segment.getMax()) {
                    return IDResult.ok(value);
                }
                if (buffer.isNextSegmentReady()) {
                    buffer.switchSegmentPosition();
                    log.info("标签 {} 已切换到缓冲区中的新号码段进行发号。", buffer.getTagName());
                    buffer.setNextSegmentReady(false);
                } else {
                    log.error("当前使用中的号码段已没有号码，且另外一个号码段还没有加载完成，无法发号。");
                    return IDResult.error(IDResult.NO_SEGMENT_AVAILABLE);
                }
            } finally {
                buffer.wLock().unlock();
            }
        }
    }

    /**
     * 自旋等待加载另一个号码段的异步任务完成。自多自旋 10000 次，并在等待 10 毫秒后放弃等待。
     * @param buffer 号码段缓冲区
     */
    private void waitAsyncLoadSegmentTaskFinish(SegmentBuffer buffer) {
        int roll = 0;
        while (buffer.getLoadSegmentThreadRunning().get()) {
            roll += 1;
            if (roll > 10000) {
                try {
                    TimeUnit.MILLISECONDS.sleep(10);
                    break;
                } catch (InterruptedException e) {
                    log.error("自旋等待号码段加载完成异常", e);
                    break;
                }
            }
        }
    }
}
